package sink;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

/**
 * 自定义Sink 实现SinkFunction
 */
public class CustomSinkDemo1 {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

        DataStreamSource<String> data = env.socketTextStream("localhost", 8888);
        // 创建匿名内部类
        SinkFunction<String> mySink = new SinkFunction<String>() {
            @Override
            public void invoke(String value, Context context) throws Exception {
                System.out.println(value);
            }
        };
        // 设置sink名称
        data.addSink(mySink).name("MyPrint");


        env.execute("Custom Sink");
    }
}
